package com.code.soulkafka.kafka;

import com.code.soulkafka.domain.NodeContext;
import com.google.common.collect.HashMultimap;
import com.google.common.collect.Multimap;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.core.task.SimpleAsyncTaskExecutor;
import org.springframework.stereotype.Component;

import java.util.*;

/**
 * @author jdy
 * @date 2:41 下午
 **/
@Slf4j
@Component
public class KafkaConsumerManager implements ApplicationContextAware {
    ApplicationContext applicationContext;

    @Autowired
    private KafkaProperties kafkaProperties;

    @Autowired
    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
    }

    private volatile boolean running;

    private Multimap<String, SimpleKafkaConsumerContainer> containerMap = HashMultimap.create();

    private SimpleAsyncTaskExecutor taskExecutor = new SimpleAsyncTaskExecutor("-KAFKA-CONSUMER-");

    public void handler(NodeContext nodeContext) {
        synchronized (containerMap) {
            if (running) {
                return;
            }
            try {
                handler0(nodeContext);
            } finally {
                running = false;
            }
        }
    }

    private void handler0(NodeContext nodeContext) {
        List<String> myTopics = nodeContext.getMyTopics();
        //判断需要删除的情况
        stopConsumer(myTopics);
        Map<String, String> idc2Server = Optional.ofNullable(nodeContext.getIdc2Server()).orElse(new HashMap<>(1));
        for (String topicKey : myTopics) {
            //已经存在
            if (containerMap.containsKey(topicKey)) {
                continue;
            }
            startConsumer(nodeContext, idc2Server, topicKey);
        }
    }

    private void stopConsumer(List<String> myTopics) {
        for (String topicKey : containerMap.keySet()) {
            if (!myTopics.contains(topicKey)) {
                //停止
                stop(topicKey);
            }
        }
    }

    private void startConsumer(NodeContext nodeContext, Map<String, String> idc2Server, String topicKey) {
        String[] topicInfo = StringUtils.split(topicKey, ":");
        int concurrency = Integer.parseInt(topicInfo[3]);
        //topic
        String[] topics = StringUtils.split(topicInfo[1], "&");
        //处理名称
        String[] handleNames = StringUtils.split(topicInfo[2], "&");
        //是否设置到从最后订阅
        String[] seekToEnds = StringUtils.split(topicInfo[4], "&");
        //是否使用相同的group
        String[] needOneGroups = StringUtils.split(topicInfo[5], "&");
        String idc = topicInfo[0];
        String server = idc2Server.get(idc);
        for (int ind = 0; ind < topics.length; ind++) {
            List<String> topic = Arrays.asList(StringUtils.split(topics[ind], "|"));
            String handleName = handleNames[ind];
            boolean needOneGroup = StringUtils.equals("1", needOneGroups[ind]);
            boolean seekToEnd = StringUtils.equals("1", seekToEnds[ind]);
            for (int i = 0; i < concurrency; i++) {
                Map<String, Object> kafkaConfig = toKafkaConfig(idc, nodeContext, server, handleName, needOneGroup);
                SimpleKafkaConsumerContainer kafkaConsumerContainer = new SimpleKafkaConsumerContainer(topic,
                        handleName, kafkaConfig, seekToEnd);
                taskExecutor.submitListenable(kafkaConsumerContainer);
                containerMap.put(topicKey, kafkaConsumerContainer);
            }
            log.info("启动消费, topic:{}, handleName:{}, concurrency:{}, seekToEnd: {}, needOneGroup:{}");
        }
    }

    private Map<String, Object> toKafkaConfig(String idc, NodeContext nodeContext, String server, String handleName, boolean needOneGroup) {
        Map<String, Object> kafkaConfig = kafkaProperties.buildConsumerProperties();
        long myId = nodeContext.getMyId();
        String suf = (needOneGroup ? StringUtils.EMPTY : "-" + myId) + "-" + idc;
        if (kafkaConfig.containsKey(ConsumerConfig.GROUP_ID_CONFIG)) {
            kafkaConfig.put(ConsumerConfig.GROUP_ID_CONFIG, kafkaConfig.get(ConsumerConfig.GROUP_ID_CONFIG) + "-" + handleName + suf);
        } else {
            kafkaConfig.put(ConsumerConfig.GROUP_ID_CONFIG, "dcsocket-" + handleName + suf);
        }
        if (StringUtils.isNotBlank(server)) {
            kafkaConfig.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, server);
        }
        return kafkaConfig;
    }

    private void stop(String topic) {
        Collection<SimpleKafkaConsumerContainer> removeList = containerMap.removeAll(topic);
        if (removeList == null) {
            return;
        }
        for (SimpleKafkaConsumerContainer container : removeList) {
            container.stop();
        }
    }
}
